package cn.gwm.flink.streaming.function.map;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @ClassName OriginalDataMapFunction
 * @Author xzh
 * @Date 2023/2/16 11:01
 **/
@Slf4j
public class OriginalData0223MapFunction implements FlatMapFunction<String, JSONObject> {

    private String vehicleType;

    private Integer minLen;

    BeanSource.SourceEnum sourceEnum;
//    List<String> vinList = Arrays.asList("LGWEEUA56NE202268","LGWEEUA58NE200831","LGWEEUA57NE201355","LGWEEUA54NE203287");

    public OriginalData0223MapFunction(String vehicleType, Integer minLen){
        this.vehicleType = vehicleType;
        this.minLen = minLen;
    }

    public OriginalData0223MapFunction(String vehicleType, Integer minLen, BeanSource.SourceEnum sourceEnum){
        this.vehicleType = vehicleType;
        this.minLen = minLen;
        this.sourceEnum = sourceEnum;
    }

    @Override
    public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
        JSONObject object = null;
        try{
            object = JSONUtil.parseObj(s);
        }catch (Exception e){
            log.error("源数据转json异常，源数据格式有误！");
            e.printStackTrace();
            throw e;
        }
        if(object.size() > minLen) {
            try {
                object.set(BaseFields.STORE_TIME, System.currentTimeMillis())
                        .set(VehicleType.VEHICLE_TYPE, vehicleType)
                        .set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                                DatePattern.NORM_DATE_PATTERN));
            } catch (Exception e) {
                log.error("json字段不存在item_time字段异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
                e.printStackTrace();
                throw e;
            }
            if (sourceEnum != null){
                try {
                    List<String> list = BeanSource.sourceList(sourceEnum);
                    JSONObject resultOjb = JSONUtil.createObj();
                    for(String feild:list){
                        resultOjb.set(feild,object.getStr(feild));
                    }
                    if("2023-02-23".equals(resultOjb.getStr(BaseFields.PARTITION_KEY))){
                        collector.collect(resultOjb);
                    }
                }catch (Exception e){
                    log.error("缩减字段异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
                    e.printStackTrace();
                    throw e;
                }
            }else{
                if("2023-02-23".equals(object.getStr(BaseFields.PARTITION_KEY))){
                    collector.collect(object);
                }
            }
        }
    }
}
